{
 "nbformat": 4,
 "nbformat_minor": 0,
 "metadata": {
  "colab": {
   "name": "CS246 - Colab 3.ipynb",
   "provenance": [],
   "collapsed_sections": [],
   "toc_visible": true
  },
  "kernelspec": {
   "name": "python3",
   "language": "python",
   "display_name": "Python 3"
  },
  "pycharm": {
   "stem_cell": {
    "cell_type": "raw",
    "source": [],
    "metadata": {
     "collapsed": false
    }
   }
  }
 },
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "kPt5q27L5557",
    "colab_type": "text"
   },
   "source": [
    "# CS246 - Colab 3\n",
    "## K-Means & PCA\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "twk-K-jilWK7",
    "colab_type": "code",
    "colab": {},
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline\n",
    "\n",
    "import pyspark\n",
    "from pyspark.sql import *\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark import SparkContext, SparkConf"
   ],
   "execution_count": 14,
   "outputs": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "BtrJlMBt1Ela"
   },
   "source": [
    "Let's initialize the Spark context."
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "colab_type": "code",
    "id": "Vm3sAVeK1EDZ",
    "colab": {},
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "# create the session\n",
    "conf = SparkConf().set(\"spark.ui.port\", \"4050\")\n",
    "\n",
    "# create the context\n",
    "sc = pyspark.SparkContext(conf=conf)\n",
    "spark = SparkSession.builder.getOrCreate()\n"
   ],
   "execution_count": 15,
   "outputs": [
    {
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-15-079f26018765>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      4\u001b[0m \u001b[0;31m# create the context\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 5\u001b[0;31m \u001b[0msc\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mpyspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mSparkContext\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mconf\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mconf\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      6\u001b[0m \u001b[0mspark\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mSparkSession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mbuilder\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgetOrCreate\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      7\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/spark-2.4.4-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/context.py\u001b[0m in \u001b[0;36m__init__\u001b[0;34m(self, master, appName, sparkHome, pyFiles, environment, batchSize, serializer, conf, gateway, jsc, profiler_cls)\u001b[0m\n\u001b[1;32m    131\u001b[0m                     \" note this option will be removed in Spark 3.0\")\n\u001b[1;32m    132\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 133\u001b[0;31m         \u001b[0mSparkContext\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_ensure_initialized\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mgateway\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mgateway\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mconf\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mconf\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    134\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    135\u001b[0m             self._do_init(master, appName, sparkHome, pyFiles, environment, batchSize, serializer,\n",
      "\u001b[0;32m~/spark-2.4.4-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/context.py\u001b[0m in \u001b[0;36m_ensure_initialized\u001b[0;34m(cls, instance, gateway, conf)\u001b[0m\n\u001b[1;32m    330\u001b[0m                         \u001b[0;34m\" created by %s at %s:%s \"\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    331\u001b[0m                         % (currentAppName, currentMaster,\n\u001b[0;32m--> 332\u001b[0;31m                             callsite.function, callsite.file, callsite.linenum))\n\u001b[0m\u001b[1;32m    333\u001b[0m                 \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    334\u001b[0m                     \u001b[0mSparkContext\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_active_spark_context\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0minstance\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mValueError\u001b[0m: Cannot run multiple SparkContexts at once; existing SparkContext(app=pyspark-shell, master=local[*]) created by __init__ at <ipython-input-3-03ad07eeace4>:5 "
     ],
     "ename": "ValueError",
     "evalue": "Cannot run multiple SparkContexts at once; existing SparkContext(app=pyspark-shell, master=local[*]) created by __init__ at <ipython-input-3-03ad07eeace4>:5 ",
     "output_type": "error"
    }
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Data Preprocessing"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "outputs": [],
   "source": [
    "from sklearn.datasets import load_breast_cancer\n",
    "breast_cancer = load_breast_cancer()"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n",
     "is_executing": false
    }
   }
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "wpsaYOqRxar2",
    "colab_type": "text"
   },
   "source": [
    "For convenience, given that the dataset is small, we first construct a Pandas dataframe, tune the schema, and then convert it into a Spark dataframe."
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "Oitav_xhQD9w",
    "colab_type": "code",
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 595
    },
    "outputId": "cbaed91c-e293-4d43-f776-2e2a3078a8f1",
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "pd_df = pd.DataFrame(breast_cancer.data, columns=breast_cancer.feature_names)\n",
    "df = spark.createDataFrame(pd_df)\n",
    "\n",
    "def set_df_columns_nullable(spark, df, column_list, nullable=False):\n",
    "    for struct_field in df.schema:\n",
    "        if struct_field.name in column_list:\n",
    "            struct_field.nullable = nullable\n",
    "    df_mod = spark.createDataFrame(df.rdd, df.schema)\n",
    "    return df_mod\n",
    "\n",
    "df = set_df_columns_nullable(spark, df, df.columns)\n",
    "df = df.withColumn('features', array(df.columns))\n",
    "vectors = df.rdd.map(lambda row: Vectors.dense(row.features))\n",
    "\n",
    "df.printSchema()"
   ],
   "execution_count": 17,
   "outputs": [
    {
     "name": "stdout",
     "text": [
      "root\n |-- mean radius: double (nullable = false)\n |-- mean texture: double (nullable = false)\n |-- mean perimeter: double (nullable = false)\n |-- mean area: double (nullable = false)\n |-- mean smoothness: double (nullable = false)\n |-- mean compactness: double (nullable = false)\n |-- mean concavity: double (nullable = false)\n |-- mean concave points: double (nullable = false)\n |-- mean symmetry: double (nullable = false)\n |-- mean fractal dimension: double (nullable = false)\n |-- radius error: double (nullable = false)\n |-- texture error: double (nullable = false)\n |-- perimeter error: double (nullable = false)\n |-- area error: double (nullable = false)\n |-- smoothness error: double (nullable = false)\n |-- compactness error: double (nullable = false)\n |-- concavity error: double (nullable = false)\n |-- concave points error: double (nullable = false)\n |-- symmetry error: double (nullable = false)\n |-- fractal dimension error: double (nullable = false)\n |-- worst radius: double (nullable = false)\n |-- worst texture: double (nullable = false)\n |-- worst perimeter: double (nullable = false)\n |-- worst area: double (nullable = false)\n |-- worst smoothness: double (nullable = false)\n |-- worst compactness: double (nullable = false)\n |-- worst concavity: double (nullable = false)\n |-- worst concave points: double (nullable = false)\n |-- worst symmetry: double (nullable = false)\n |-- worst fractal dimension: double (nullable = false)\n |-- features: array (nullable = false)\n |    |-- element: double (containsNull = false)\n\n"
     ],
     "output_type": "stream"
    }
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "TtR1xRvonxiO",
    "colab_type": "text"
   },
   "source": [
    "With the next cell, we build the two datastructures that we will be using throughout this Colab:\n",
    "\n",
    "\n",
    "*   ```features```, a dataframe of Dense vectors, containing all the original features in the dataset;\n",
    "*   ```labels```, a series of binary labels indicating if the corresponding set of features belongs to a subject with breast cancer, or not.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "GP23Xkgwi0SD",
    "colab_type": "code",
    "colab": {},
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "from pyspark.ml.linalg import Vectors\n",
    "features = spark.createDataFrame(vectors.map(Row), [\"features\"])\n",
    "labels = pd.Series(breast_cancer.target)"
   ],
   "execution_count": 18,
   "outputs": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "CRaF2A_j_nC7",
    "colab_type": "text"
   },
   "source": [
    "### Your task"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ebLNUxP0_8x3",
    "colab_type": "text"
   },
   "source": [
    "If you run successfully the Setup and Data Preprocessing stages, you are now ready to cluster the data with the [K-means](https://spark.apache.org/docs/latest/ml-clustering.html) algorithm included in MLlib (Spark's Machine Learning library).\n",
    "Set the ```k``` parameter to **2**, fit the model, and the compute the [Silhouette score](https://en.wikipedia.org/wiki/Silhouette_(clustering)) (i.e., a measure of quality of the obtained clustering).  \n",
    "\n",
    "**IMPORTANT:** use the MLlib implementation of the Silhouette score (via ```ClusteringEvaluator```)."
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "0xVIfPHZwWaE",
    "colab_type": "code",
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "outputId": "20a5940b-30e6-4d65-89b2-7de97231c1b6",
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "# YOUR CODE HERE\n",
    "from pyspark.ml.clustering import KMeans\n",
    "from pyspark.ml.evaluation import ClusteringEvaluator\n",
    "\n",
    "# Trains a k-means model.\n",
    "kmeans = KMeans().setK(2).setSeed(1)\n",
    "model = kmeans.fit(features)\n",
    "\n",
    "# Make predictions\n",
    "predictions = model.transform(features)\n",
    "# Evaluate clustering by computing Silhouette score\n",
    "evaluator = ClusteringEvaluator()\n",
    "\n",
    "silhouette = evaluator.evaluate(predictions)\n",
    "print(\"Silhouette with squared euclidean distance = \" + str(silhouette))"
   ],
   "execution_count": 20,
   "outputs": [
    {
     "name": "stdout",
     "text": [
      "Silhouette with squared euclidean distance = 0.8342904262826143\n"
     ],
     "output_type": "stream"
    }
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "5GB09n7sqTO6",
    "colab_type": "text"
   },
   "source": [
    "Take the predictions produced by K-means, and compare them with the ```labels``` variable (i.e., the ground truth from our dataset).  \n",
    "\n",
    "Compute how many data points in the dataset have been clustered correctly (i.e., positive cases in one cluster, negative cases in the other).\n",
    "\n",
    "*HINT*: you can use ```np.count_nonzero(series_a == series_b)``` to quickly compute the element-wise comparison of two series.\n",
    "\n",
    "**IMPORTANT**: K-means is a clustering algorithm, so it will not output a label for each data point, but just a cluster identifier!  As such, label ```0``` does not necessarily match the cluster identifier ```0```.\n"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "WQhC3APIPPM5",
    "colab_type": "code",
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "outputId": "7d4f5570-90d3-48ff-9129-96114eaa2438",
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "# YOUR CODE HERE\n",
    "predictions_series=predictions.select('prediction').toPandas()\n",
    "true_prediction_count=np.count_nonzero(predictions_series['prediction']==labels)\n",
    "if true_prediction_count<len(labels)-true_prediction_count:\n",
    "  true_prediction_count=len(labels)-true_prediction_count\n",
    "print('precision:',true_prediction_count/len(labels))\n"
   ],
   "execution_count": 22,
   "outputs": [
    {
     "name": "stdout",
     "text": [
      "precision: 0.8541300527240774\n"
     ],
     "output_type": "stream"
    }
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "fLIprM1JsdTU",
    "colab_type": "text"
   },
   "source": [
    "Now perform dimensionality reduction on the ```features``` using the [PCA](https://spark.apache.org/docs/latest/ml-features.html#pca) statistical procedure, available as well in MLlib.\n",
    "\n",
    "Set the ```k``` parameter to **2**, effectively reducing the dataset size of a **15X** factor."
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "p4J8JMDkSb24",
    "colab_type": "code",
    "colab": {},
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "# YOUR CODE HERE\n",
    "from pyspark.ml.feature import PCA\n",
    "from pyspark.ml.linalg import Vectors\n",
    "\n",
    "\n",
    "pca = PCA(k=2, inputCol=\"features\", outputCol=\"pcaFeatures\")\n",
    "pca_model = pca.fit(features)\n",
    "pcaFeatures = pca_model.transform(features).select(\"pcaFeatures\")"
   ],
   "execution_count": 23,
   "outputs": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "8leQR4-atMAl",
    "colab_type": "text"
   },
   "source": [
    "Now run K-means with the same parameters as above, but on the ```pcaFeatures``` produced by the PCA reduction you just executed.\n",
    "\n",
    "Compute the Silhouette score, as well as the number of data points that have been clustered correctly."
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "U_snSSj5k2y5",
    "colab_type": "code",
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "outputId": "a1d50595-e975-4092-fd9b-5ae828c091fe",
    "pycharm": {
     "is_executing": false
    }
   },
   "source": [
    "# YOUR CODE HERE\n",
    "kmeans = KMeans(featuresCol='pcaFeatures').setK(2).setSeed(1)\n",
    "pca_kmeans_model = kmeans.fit(pcaFeatures)\n",
    "# Make predictions\n",
    "pca_predictions = pca_kmeans_model.transform(pcaFeatures)\n",
    "# Evaluate clustering by computing Silhouette score\n",
    "evaluator = ClusteringEvaluator()\n",
    "\n",
    "silhouette = evaluator.evaluate(pca_predictions.withColumnRenamed('pcaFeatures','features'))\n",
    "print(\"Silhouette with squared euclidean distance = \" + str(silhouette))"
   ],
   "execution_count": 24,
   "outputs": [
    {
     "name": "stdout",
     "text": [
      "Silhouette with squared euclidean distance = 0.8348610363444835\n"
     ],
     "output_type": "stream"
    }
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "sMUb_afmlIEW",
    "colab_type": "code",
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "outputId": "81c7d83d-7b94-4079-ec55-6c781ffa5a21",
    "pycharm": {
     "is_executing": false,
     "name": "#%%\n"
    }
   },
   "source": [
    "# YOUR CODE HERE\n",
    "pca_predictions_series=pca_predictions.select('prediction').toPandas()\n",
    "true_prediction_count=np.count_nonzero(pca_predictions_series['prediction']==labels)\n",
    "if true_prediction_count<len(labels)-true_prediction_count:\n",
    "  true_prediction_count=len(labels)-true_prediction_count\n",
    "print('precision:',true_prediction_count/len(labels))\n"
   ],
   "execution_count": 25,
   "outputs": [
    {
     "name": "stdout",
     "text": [
      "precision: 0.8541300527240774\n"
     ],
     "output_type": "stream"
    }
   ]
  }
 ]
}